package org.eleorc.flow;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;
import java.util.concurrent.atomic.AtomicInteger;

public class FlowDemo02 {

    public static class MySubscriber implements Flow.Subscriber<Integer> {

        private int speed;

        private Flow.Subscription subscription;

        public MySubscriber(int speed) {
            this.speed = speed;
        }

        @Override
        public void onSubscribe(Flow.Subscription subscription) {
            this.subscription = subscription;
            subscription.request(1);
        }

        @Override
        public void onNext(Integer item) {
            System.out.println(Thread.currentThread().getName()+"=>"+item);
            subscription.request(1);
            try {
                Thread.sleep(speed);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        @Override
        public void onError(Throwable throwable) {

        }

        @Override
        public void onComplete() {
            System.out.println(Thread.currentThread().getName()+":onComplete");
        }
    }

    public static void main(String[] args) throws InterruptedException {
        SubmissionPublisher<Integer> publisher = new SubmissionPublisher<>();
        publisher.subscribe(new MySubscriber(200));
        publisher.subscribe(new MySubscriber(300));

        for (int i = 0; i < 10; i++) {
            publisher.submit(i);
        }
        publisher.close();
        Thread.sleep(5000);
        System.out.println("main thread exit");
        System.exit(0);


    }

}
